package com.shanks.framework.mq.data.mybatis.repository;


import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.shanks.framework.mq.data.mybatis.mapper.MessageRecordMapper;
import com.shanks.framework.mq.data.mybatis.service.MessageRecordService;
import com.shanks.framework.mq.enums.MessageRecordStatusEnum;
import com.shanks.framework.mq.model.MessageRecordPO;
import com.shanks.framework.mq.repository.MessageRecordRepository;
import jakarta.annotation.Resource;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.Objects;

@Service
public class MessageRecordRepositoryImpl implements MessageRecordRepository {
    @Resource
    private MessageRecordMapper messageRecordMapper;

    @Resource
    private MessageRecordService messageRecordService;

    @Override
    public Boolean updateMessageStatus(Long messageId, MessageRecordStatusEnum oldMessageStatus, MessageRecordStatusEnum newMessageStatus) {
        LambdaUpdateWrapper<MessageRecordPO> lambdaUpdateWrapper = bulidUpdateWrapper(oldMessageStatus, newMessageStatus);
        lambdaUpdateWrapper.eq(MessageRecordPO::getMessageId, messageId);
        return messageRecordMapper.update(lambdaUpdateWrapper) > 0;
    }

    private LambdaUpdateWrapper<MessageRecordPO> bulidUpdateWrapper(MessageRecordStatusEnum oldMessageStatus, MessageRecordStatusEnum newMessageStatus) {
        LambdaUpdateWrapper<MessageRecordPO> lambdaUpdateWrapper = Wrappers.lambdaUpdate(MessageRecordPO.class)
                .isNull(MessageRecordPO::getAckTime);
        if (Objects.nonNull(oldMessageStatus)) {
            lambdaUpdateWrapper.eq(MessageRecordPO::getMessageRecordStatus, oldMessageStatus);
        }
        if (newMessageStatus == MessageRecordStatusEnum.SENDING) {
            lambdaUpdateWrapper.set(MessageRecordPO::getSendingTime, new Date());
            lambdaUpdateWrapper.setSql("retry_count = retry_count + 1");
        } else if (newMessageStatus == MessageRecordStatusEnum.SENT) {
            lambdaUpdateWrapper.set(MessageRecordPO::getSentTime, new Date());
        } else if (newMessageStatus == MessageRecordStatusEnum.ACKED) {
            lambdaUpdateWrapper.set(MessageRecordPO::getAckTime, new Date());
        }
        lambdaUpdateWrapper.set(MessageRecordPO::getMessageRecordStatus, newMessageStatus);
        return lambdaUpdateWrapper;
    }

    @Override
    public void updateMessageRetryCount(Long messageId, int retryCount) {
        MessageRecordPO updatePO = new MessageRecordPO();
        updatePO.setMessageId(messageId);
        updatePO.setRetryCount(retryCount);
        messageRecordMapper.updateById(updatePO);
    }

    @Override
    public Boolean updateMessageStatusByUuid(String messageUuid, MessageRecordStatusEnum oldMessageStatus, MessageRecordStatusEnum newMessageStatus, Date updateTime, String consumeClientSource) {
        LambdaUpdateWrapper<MessageRecordPO> lambdaUpdateWrapper = bulidUpdateWrapper(oldMessageStatus, newMessageStatus);
        lambdaUpdateWrapper.eq(MessageRecordPO::getMessageUuid, messageUuid);
        return messageRecordMapper.update(lambdaUpdateWrapper) > 0;
    }

    @Override
    public void insert(MessageRecordPO messageRecordPO) {
        messageRecordMapper.insert(messageRecordPO);
    }

    @Override
    public void insertBatch(List<MessageRecordPO> messageRecordPOList) {
        messageRecordService.saveBatch(messageRecordPOList);
    }

    @Override
    public List<MessageRecordPO> selectPendingMessageList(Date createTimeStart, Date createTimeEnd, int pageSize, int pageIndex, List<String> producerKeys) {
        IPage<MessageRecordPO> page = new Page(pageIndex, pageSize);
        LambdaQueryWrapper<MessageRecordPO> queryWrapper = buildQueryWrapper(createTimeStart, producerKeys);
        queryWrapper.eq(MessageRecordPO::getMessageRecordStatus, MessageRecordStatusEnum.PENDING);
        queryWrapper.lt(MessageRecordPO::getCreateTime, createTimeEnd);
        return messageRecordMapper.selectPage(page, queryWrapper).getRecords();
    }

    private LambdaQueryWrapper<MessageRecordPO> buildQueryWrapper(Date createTimeStart, List<String> producerKeys) {
        LambdaQueryWrapper<MessageRecordPO> queryWrapper = Wrappers.lambdaQuery(MessageRecordPO.class)
                .gt(MessageRecordPO::getCreateTime, createTimeStart);
        if (!CollectionUtils.isEmpty(producerKeys)) {
            queryWrapper.in(MessageRecordPO::getProducerKey, producerKeys);
        }
        queryWrapper.last("ORDER BY retry_count, message_id ASC, create_time");
        return queryWrapper;
    }

    @Override
    public List<MessageRecordPO> selectSendTimeoutMessageList(Date createTimeStart, Date sendingTimeEnd, int pageSize, int pageIndex, List<String> producerKeys) {
        IPage<MessageRecordPO> page = new Page(pageIndex, pageSize);
        LambdaQueryWrapper<MessageRecordPO> queryWrapper = buildQueryWrapper(createTimeStart, producerKeys);
        queryWrapper.eq(MessageRecordPO::getMessageRecordStatus, MessageRecordStatusEnum.SENDING);
        queryWrapper.lt(MessageRecordPO::getSendingTime, sendingTimeEnd);
        return messageRecordMapper.selectPage(page, queryWrapper).getRecords();
    }

    @Override
    public List<MessageRecordPO> selectAckTimeoutMessageList(Date createTimeStart, Date sentTimeEnd, int pageSize, int pageIndex, List<String> producerKeys, List<String> ackCheckingTopics) {
        IPage<MessageRecordPO> page = new Page(pageIndex, pageSize);
        LambdaQueryWrapper<MessageRecordPO> queryWrapper = buildQueryWrapper(createTimeStart, producerKeys);
        queryWrapper.eq(MessageRecordPO::getMessageRecordStatus, MessageRecordStatusEnum.SENT);
        queryWrapper.lt(MessageRecordPO::getSendingTime, sentTimeEnd);
        if (!CollectionUtils.isEmpty(ackCheckingTopics)) {
            queryWrapper.notIn(MessageRecordPO::getProducerSource, ackCheckingTopics);
        }
        IPage<MessageRecordPO> messageRecordPOIPage = messageRecordMapper.selectPage(page, queryWrapper);
        return messageRecordPOIPage.getRecords();
    }
}
